/*
 * Middleware
 * Copyright (C) 2010/11
 * Riccardo Pulvirenti, Giuseppe Ravidà, Andrea Tino
 *
 * This program is free software: you can redistribute it and/or modify
 * it under the terms of the GNU General Public License as published by
 * the Free Software Foundation, either version 3 of the License, or
 * (at your option) any later version.
 *
 * This program is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANDABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
 * GNU General Public License for more details.
 *
 * You should have received a copy of the GNU General Public License
 * along with this program. If not, see http://www.gnu.org/licenses
 */

/*!
 * File: worker_discovery.cpp
 */

#include <iostream>
#include <string>
#include <exception>
#include <boost/asio.hpp>
#include <boost/lexical_cast.hpp>
#include <boost/archive/binary_iarchive.hpp>
#include <boost/archive/binary_oarchive.hpp>
#include <boost/thread/mutex.hpp>
#include <cstdlib>

#include "worker_discovery.hpp"
#include "utils.hpp"

using namespace middleware;

//-----------------------------------------------------------------------------
// Constructors and destructor
/* Default constructor */
WorkerDiscovery::WorkerDiscovery() {
	this->_config.disc_ip_toman_bind = "127.0.0.1"; /* Not used to bind */
	this->_config.disc_port_toman_bind = this->createPort();
	this->_config.disc_ip_toworkers_bind = "127.0.0.1"; /* Not used to bind */
	this->_config.disc_port_toworkers_bind = this->createPort();
}
/* Constructor with configuration */
WorkerDiscovery::WorkerDiscovery(DiscoveryConfig conf) {
	this->_config.disc_ip_toman_bind = conf.disc_ip_toman_bind;
	this->_config.disc_port_toman_bind = conf.disc_port_toman_bind;
	this->_config.disc_ip_toworkers_bind = conf.disc_ip_toworkers_bind;
	this->_config.disc_port_toworkers_bind = conf.disc_port_toworkers_bind;
}
/* Destructor */
WorkerDiscovery::~WorkerDiscovery() {
	// All members created statically.
}
//-----------------------------------------------------------------------------
// Private members
/* Gets a random port */
std::string WorkerDiscovery::createPort() {
	return utils::GetRandomIntegralNumberStr(kWorkerDiscoveryMinManPort,
			kWorkerDiscoveryMaxManPort);
}
/* Thread execution flow */
void WorkerDiscovery::exec() {
	using boost::asio::ip::tcp;
	// Execution flow for Discovery
	std::cout << kWorkerDiscoveryEtq << "Running!" << std::endl;
	// Accepting a TaskDescriptor and getting a WorkerDescriptor.
	try {
		int port = boost::lexical_cast<int>(
					this->_config.disc_port_toman_bind);
		boost::asio::io_service io_s;
		tcp::acceptor td_acceptor(io_s, tcp::endpoint(tcp::v4(), port));
		for (;;) {
			tcp::iostream* td_stream = new tcp::iostream;
			TaskDescriptor td;
			std::cout << kWorkerDiscoveryEtq <<
					"Waiting for incoming task descriptor...!" << std::endl;
			td_acceptor.accept(*(td_stream->rdbuf()));
			boost::archive::binary_iarchive ia(*td_stream);
			ia >> td; /* Here I have the task descriptor */
			std::cout << kWorkerDiscoveryEtq <<
					"Task descriptor arrived: id=" << td.get_id() << std::endl;
			// We should get a proper worker to this task
			boost::shared_ptr<boost::thread>(new boost::thread(
					boost::bind(&WorkerDiscovery::ManageTaskDescriptor,
							this, td, td_stream))); /* Thread released */
		}
	} catch(std::exception& e) {
		std::cout << kWorkerDiscoveryEtq << "Error: " << e.what() << std::endl;
		std::cout << kWorkerDiscoveryEtq << "Program now dying..." << std::endl;
		exit(1);
	}
}
/* Secondary thread execution flow */
void WorkerDiscovery::exec_worker_status() {
	using boost::asio::ip::tcp;
	std::cout << kWorkerDiscoveryEtq << kWorkerDiscovery1Etq <<
			"Running!" << std::endl;
	/* This thread must listen for incoming workerdescriptors and put them
	 * in the vector, if not present, or update it by replacing it in
	 * the vector */
	try {
		int port = boost::lexical_cast<int>(
				this->_config.disc_port_toworkers_bind);
		boost::asio::io_service io_s;
		tcp::acceptor wd_acceptor(io_s, tcp::endpoint(tcp::v4(), port));
		for (;;) {
			tcp::iostream wd_stream;
			WorkerDescriptor wd;
			std::cout << kWorkerDiscoveryEtq << kWorkerDiscovery1Etq <<
					"Waiting for incoming worker descriptor...!" <<	std::endl;
			wd_acceptor.accept(*(wd_stream.rdbuf()));
			boost::archive::binary_iarchive ia(wd_stream);
			ia >> wd; /* Here I have the worker descriptor */
			wd_stream.close();
			std::cout << kWorkerDiscoveryEtq << kWorkerDiscovery1Etq <<
					"Worker descriptor arrived: address(ip=" <<
					wd.get_address().get_ip() << ", port=" <<
					wd.get_address().get_port() << ")" << std::endl;
			this->ManageWorkerDescriptor(wd);
		}
	} catch(std::exception& e) {
		std::cout << kWorkerDiscoveryEtq << kWorkerDiscovery1Etq <<
				"Error: " << e.what() << std::endl;
		std::cout << kWorkerDiscoveryEtq << kWorkerDiscovery1Etq <<
				"Program now dying..." << std::endl;
		exit(1);
	}
}
/* Thread task manager */
void WorkerDiscovery::ManageTaskDescriptor(TaskDescriptor td,
		boost::asio::ip::tcp::iostream* iostrm) {
	WorkerDescriptor wd = this->GetCurrentBestWorkerDescriptor();
	std::cout << kWorkerDiscoveryEtq << kWorkerDiscovery1Etq << "Thread: " <<
			"Selected worker: ip=" << wd.get_address().get_ip() << " PI=" <<
			((double)wd.PI()) << std::endl;
	// Send back the worker descriptor
	// TODO
	// Clearing stream after closing it
	iostrm->close();
	delete iostrm;
}
/* Getting current best worker (worker descriptor), LOCKING CALL */
WorkerDescriptor WorkerDiscovery::GetCurrentBestWorkerDescriptor() {
	// Cycle in worker descriptor getting the one having the highest PI
	boost::mutex::scoped_lock lck(this->_workerdescs_monitor); /* Lock */
	std::vector<WorkerDescriptor>::iterator it;
	std::vector<WorkerDescriptor>::iterator highest_it;
	calculus::PIType highest_pi = 0;
	for (it = this->_workerdescs.begin(); it < this->_workerdescs.end();
			it++) {
		if (it->PI() > highest_pi) {
			highest_pi = it->PI();
			highest_it = it;
		}
	}
	lck.unlock(); /* Unlock */
	// Here I have the final value
	return (WorkerDescriptor)(*highest_it);
}
/* Manage a worker descriptor */
void WorkerDiscovery::ManageWorkerDescriptor(WorkerDescriptor wd) {
	/* Consider the worker descriptor, if it is not in the vector, add it, if
	 * its address appears in the list, replace it */
	std::vector<WorkerDescriptor>::iterator it;
	boost::mutex::scoped_lock lck(this->_workerdescs_monitor); /* Lock */
	for (it = this->_workerdescs.begin(); it < this->_workerdescs.end();
			it++) {
		if (it->get_address() == wd.get_address()) {
			it->set_capabilities(wd.get_capabilities());
			it->set_status(wd.get_status());
			lck.unlock(); /* Unlock */
			return; /* Ends here */
		}
	}
	/* If arrives here, no wdesc was found, this is a registration for a new
	 * worker. */
	this->_workerdescs.push_back(wd);
	lck.unlock(); /* Unlock */
}
/* Bonjour */
void WorkerDiscovery::bonjour() {
	std::cout << kWorkerDiscoveryEtq <<
			"Initialization sequence now started..." <<	std::endl;
	std::cout << kWorkerDiscoveryEtq <<
			"Configuration loaded: bind_tomanIP: this" <<
			", bind_tomanPORT: " <<
			this->_config.disc_port_toman_bind <<
			"; bind_toworkersIP: this" << ", bind_toworkersPORT: " <<
			this->_config.disc_port_toworkers_bind << std::endl;
	std::cout << kWorkerDiscoveryEtq << "Initialization completed!" <<
			std::endl;
}
